import asyncio
import json
import random
from dataclasses import dataclass
from typing import Optional, List, Dict, Any

from tcp_server.tcp_clients import ClientManager
from utils.log_util import logger


@dataclass
class TaskConfig:
    keyword: str
    total_time: int = 60              # 一轮总用时（秒）
    enable_auto_order: bool = False
    skip_first_search: bool = False    # ✅ 首次搜索不自动下单
    enable_jitter: bool = True         # ✅ 抖动搜索（默认开启，模拟真人行为）
    order_price: float = 0.0          # 0=有新就下
    min_price: float = 0.0            # 0/None 不限制
    max_price: float = 0.0            # 0/None 不限制
    display_count: int = 30           # 前端显示数量，不影响调度
    user_id: Optional[int] = None     # ✅ 新增：用户ID（用于多用户隔离）
    task_id: Optional[int] = None     # ✅ 新增：任务ID（用于标识任务）


class SearchScheduler:
    """控制端搜索任务调度器（核心）

    - 轮转可用的搜索客户端，基于一轮总时长计算每台的时间片
    - 向 Agent 发送 `search` 指令；Agent 侧做黑名单过滤、缓存对比、自动下单
    - 通过 WebSocket 广播任务状态由调用方实现（这里聚焦调度核心）
    - ✅ 支持多用户独立调度（每个用户有独立的调度器实例）
    """

    def __init__(self, user_id: Optional[int] = None) -> None:
        self._user_id: Optional[int] = user_id  # ✅ 用户ID（用于隔离）
        self._config: Optional[TaskConfig] = None
        self._running: bool = False
        self._runner_task: Optional[asyncio.Task] = None
        self._rotation_index: int = 0
        self._round_count: int = 0  # ✅ 轮次计数

    def is_running(self) -> bool:
        return self._running

    def get_config(self) -> Optional[TaskConfig]:
        return self._config
    
    def get_round_count(self) -> int:
        """获取当前轮次计数"""
        return self._round_count

    async def get_current_client_id(self) -> Optional[str]:
        """获取当前正在搜索的客户端ID（只返回search类型的客户端）"""
        if not self._running:
            return None
        try:
            # ✅ 使用 await 调用异步方法
            clients = await self._get_search_clients()
            # ✅ _get_search_clients() 已经确保只返回 search 类型的客户端
            # 所以这里可以直接使用，不需要再次验证
            if clients and self._rotation_index > 0:
                # 计算当前客户端（基于轮转索引，索引已经递增，所以减1）
                current_index = (self._rotation_index - 1) % len(clients) if clients else 0
                if current_index < len(clients):
                    return clients[current_index].client_id
        except Exception as e:
            logger.error(f"获取当前搜索客户端ID失败: {e}", exc_info=True)
        return None

    async def start(self, cfg: TaskConfig) -> None:
        if self._running:
            logger.info(f"SearchScheduler 已在运行，更新配置并继续（关键词={cfg.keyword}）")
            self._config = cfg
            return
        self._config = cfg
        self._running = True
        self._rotation_index = 0  # ✅ 重置轮转索引
        self._round_count = 0  # ✅ 重置轮次计数
        self._runner_task = asyncio.create_task(self._run_loop())
        jitter_mode = "抖动搜索" if cfg.enable_jitter else "平均搜索"
        logger.info(
            f"SearchScheduler 启动成功（关键词={cfg.keyword}，总时长={cfg.total_time}秒，"
            f"搜索模式={jitter_mode}，自动下单={cfg.enable_auto_order}）"
        )

    async def stop(self) -> None:
        self._running = False
        if self._runner_task:
            self._runner_task.cancel()
            try:
                await self._runner_task
            except asyncio.CancelledError:
                pass
            finally:
                self._runner_task = None
        logger.info("SearchScheduler 已停止")

    async def _run_loop(self) -> None:
        logger.info(f"SearchScheduler: 运行循环已启动，关键词={self._config.keyword}")
        while self._running:
            try:
                clients = await self._get_search_clients()
                if not clients:
                    logger.warning(f"SearchScheduler: 没有可用的搜索客户端，等待2秒后重试...")
                    await asyncio.sleep(2)
                    continue

                logger.debug(f"SearchScheduler: 找到 {len(clients)} 个可用搜索客户端，开始分发搜索任务")
                
                # 计算每台客户端的基础时间片（至少1秒）
                total_time = max(int(self._config.total_time or 60), 1)
                base_slice = max(total_time // len(clients), 1)
                
                # ✅ 根据配置决定是否使用随机抖动
                if self._config.enable_jitter:
                    # 抖动搜索：混合策略，固定抖动 ± 比例抖动
                    # 固定抖动：±1秒（短间隔时的基础抖动）
                    fixed_jitter = random.uniform(-1.0, 1.0)
                    # 比例抖动：±15%（长间隔时更自然）
                    ratio_jitter = base_slice * random.uniform(-0.15, 0.15)
                    total_jitter = fixed_jitter + ratio_jitter
                    
                    # 实际时间片（最小0.5秒，避免过快）
                    actual_slice = max(base_slice + total_jitter, 0.5)
                else:
                    # 平均搜索：不使用抖动，严格按固定时间片
                    total_jitter = 0.0
                    actual_slice = base_slice

                # ✅ 每轮开始时增加轮次计数
                # 一轮 = 所有客户端都搜索完一次
                # 当rotation_index为0时，表示第一轮开始（初始状态）
                # 当rotation_index能被客户端数量整除时，表示完成了一轮，开始新一轮
                if self._rotation_index % len(clients) == 0:
                    self._round_count += 1
                    logger.debug(f"SearchScheduler: 开始第 {self._round_count} 轮搜索（共 {len(clients)} 个客户端）")

                # 轮转选择
                client = clients[self._rotation_index % len(clients)]
                self._rotation_index += 1

                # ✅ 日志输出：区分抖动和平均模式
                if self._config.enable_jitter:
                    logger.debug(
                        f"SearchScheduler: 选择客户端 {client.client_id} 进行搜索 "
                        f"（关键词={self._config.keyword}，基础={base_slice}秒，"
                        f"抖动={total_jitter:+.2f}秒，实际={actual_slice:.2f}秒，"
                        f"模式=抖动搜索，轮次={self._round_count}）"
                    )
                else:
                    logger.debug(
                        f"SearchScheduler: 选择客户端 {client.client_id} 进行搜索 "
                        f"（关键词={self._config.keyword}，时间片={actual_slice:.2f}秒，"
                        f"模式=平均搜索，轮次={self._round_count}）"
                    )
                await self._dispatch_search(client.client_id)

                await asyncio.sleep(actual_slice)
            except Exception as e:
                logger.exception(f"SearchScheduler 运行异常: {e}")
                await asyncio.sleep(2)

    async def _get_search_clients(self):
        """
        获取在线且标记为 search 的客户端；
        ✅ 优先使用TCP连接状态（最准确），然后与数据库状态匹配
        ✅ 过滤掉 client_type='order' 的客户端，下单账号不参与搜索；
        ✅ 过滤掉 work_status='error' 的客户端，避免继续向异常主机发送请求；
        若无法查询数据库，则退化为所有在线客户端。
        """
        try:
            # ✅ 优先：在线 TCP 客户端 map（这是最准确的在线状态）
            online_tcp = {c.client_id: c for c in ClientManager.get_all_clients()}
            if not online_tcp:
                logger.warning("SearchScheduler: 没有TCP连接的客户端")
                return []
            
            
            # 运行时导入以避免循环依赖
            from config.get_db import get_db
            from module_admin.service.agent_client_service import AgentClientService
            
            # ✅ 修复：直接使用异步，不需要 run_until_complete
            async for db in get_db():
                # ✅ 直接从数据库获取所有 search 类型的客户端（不管 online_status）
                # 然后与TCP连接状态匹配，这样即使数据库状态不对，也能找到在线的客户端
                # ✅ 如果指定了user_id，只返回分配给该用户的搜索客户端
                clients = await AgentClientService.get_available_search_clients(
                    db, 
                    ignore_online_status=True,
                    user_id=self._user_id  # ✅ 按user_id过滤
                )
                
                # ✅ 如果数据库查询失败或返回空，回退到使用TCP连接状态
                if not clients:
                    logger.warning("SearchScheduler: 数据库中没有搜索客户端，尝试使用TCP连接状态")
                    # 直接从TCP连接中获取，但需要验证client_type
                    # 由于无法从TCP连接获取client_type，这里回退为所有TCP连接的客户端
                    # 但在_dispatch_search中会再次验证client_type
                    return list(online_tcp.values())
                
                
                # ✅ 过滤条件：
                # 1. client_type='search' 已经在 get_available_search_clients 中过滤了（只返回search类型的客户端）
                # 2. ✅ 过滤掉 work_status = 'error' 的客户端（主机异常时不再调度）
                # 3. ✅ 过滤掉 work_status = 'banned' 的客户端（被封禁的账户不能搜索）
                # 4. ✅ 过滤掉 work_status = 'searching' 的客户端（正在处理中，包括风控验证，避免并发冲突）
                # 5. ✅ 关键：同时检查TCP连接状态（优先使用TCP状态，因为这是最准确的）
                # 6. ✅ 必须已登录（login_status='logged_in'），未登录的客户端无法搜索
                # 7. ✅ 排除登录中（login_status='logging_in'）和登录失败（login_status='login_error'）的状态
                filtered_ids = [
                    c.client_id for c in clients 
                    if c.client_id in online_tcp  # ✅ 必须在TCP连接中存在（这是最准确的在线状态）
                    and (c.work_status or 'idle') != 'error'  # ✅ 排除异常状态
                    and (c.work_status or 'idle') != 'banned'  # ✅ 排除被封禁的账户
                    and (c.work_status or 'idle') != 'searching'  # ✅ 排除正在处理中的状态（包括风控验证）
                    and (c.login_status or 'not_logged_in') == 'logged_in'  # ✅ 必须已登录，未登录的客户端无法搜索
                ]
                
                # ✅ 返回TCP客户端对象（而不是数据库对象），确保连接可用
                result = [online_tcp[cid] for cid in filtered_ids if cid in online_tcp]
                logger.debug(f"SearchScheduler: 找到 {len(result)} 个可用搜索客户端")
                return result
            
            # 如果 get_db() 没有返回任何数据库会话，返回空列表
            return []
        except Exception as e:
            logger.error(f"SearchScheduler: 获取搜索客户端失败: {e}", exc_info=True)
            # ⚠️ 回退方案：如果数据库查询失败，不能返回所有TCP客户端（可能包含下单客户端）
            # 回退时只返回TCP连接的客户端，但在_dispatch_search中会验证client_type
            logger.warning(f"SearchScheduler: 使用回退方案，但会在_dispatch_search中验证client_type")
            # ⚠️ 重要：回退时返回空列表，避免使用错误的客户端类型
            # 如果确实需要回退，应该在_dispatch_search中进行严格验证
            return []

    async def _dispatch_search(self, client_id: str) -> None:
        """向指定客户端发送搜索命令
        
        注意：搜索客户端只负责搜索，不执行下单。
        自动下单逻辑由后端服务端在收到搜索结果后，选择下单客户端来执行。
        """
        # ✅ 双重检查：验证客户端类型（防止下单账号参与搜索）
        try:
            from config.get_db import get_db
            from module_admin.service.agent_client_service import AgentClientService
            async for db in get_db():
                # 获取客户端信息
                client = await AgentClientService.get_client_by_client_id(db, client_id)
                if client:
                    client_type = client.client_type if hasattr(client, 'client_type') else getattr(client, 'client_type', None)
                    if client_type and client_type != 'search':
                        return  # 直接返回，不发送搜索命令
                break
        except Exception as e:
            logger.debug(f"验证客户端类型失败（不影响搜索）: {e}")
        
        # 组装与 Agent 协议兼容的 payload
        payload: Dict[str, Any] = {
            "client_id": client_id,
            "keyword": self._config.keyword,
            # 价格区间传给 Agent，由 Agent 侧决定是否 fromFilter
            "min_price": float(self._config.min_price or 0),
            "max_price": float(self._config.max_price or 0),
            # ✅ 新增：传递user_id和task_id（用于缓存隔离和任务关联）
            "user_id": self._config.user_id,
            "task_id": self._config.task_id,
            # 注意：不再发送 auto_order 配置
            # 自动下单改为由后端服务端协调：搜索到新商品后，选择下单客户端来执行
        }

        cmd_message = {
            "event": "search",
            "direction": "req",
            "success": True,
            "code": 0,
            "msg": "",
            "data": payload,
        }

        ok = await ClientManager.send_to_client(client_id, cmd_message)
        if ok:
            logger.debug(f"SearchScheduler: 向 {client_id} 发送搜索命令: {self._config.keyword}")
        else:
            logger.error(f"SearchScheduler: ❌ 向客户端 {client_id} 发送搜索命令失败（客户端可能已断开连接）")


# ✅ 多用户调度器管理器
class SearchSchedulerManager:
    """搜索调度器管理器（支持多用户独立调度）"""
    
    def __init__(self):
        self._schedulers: Dict[int, SearchScheduler] = {}  # {user_id: SearchScheduler}
        self._lock = asyncio.Lock()
    
    async def get_scheduler(self, user_id: int) -> SearchScheduler:
        """获取指定用户的调度器实例（如果不存在则创建）"""
        async with self._lock:
            if user_id not in self._schedulers:
                self._schedulers[user_id] = SearchScheduler(user_id=user_id)
                logger.info(f"✅ 为用户 {user_id} 创建新的调度器实例")
            return self._schedulers[user_id]
    
    async def remove_scheduler(self, user_id: int):
        """移除指定用户的调度器实例（停止并清理）"""
        async with self._lock:
            if user_id in self._schedulers:
                scheduler = self._schedulers[user_id]
                if scheduler.is_running():
                    await scheduler.stop()
                del self._schedulers[user_id]
                logger.info(f"✅ 已移除用户 {user_id} 的调度器实例")
    
    async def stop_all(self):
        """停止所有调度器"""
        async with self._lock:
            for user_id, scheduler in list(self._schedulers.items()):
                if scheduler.is_running():
                    await scheduler.stop()
            self._schedulers.clear()
            logger.info("✅ 已停止所有调度器")


# ✅ 全局调度器管理器（替代原来的全局单例）
scheduler_manager = SearchSchedulerManager()

# ✅ 兼容旧代码：保留全局单例（但建议使用scheduler_manager）
search_scheduler = SearchScheduler()


